package com.ai.service.rocketMq.impl;

import com.ai.common.constant.enumP.MqMessageStatus;
import com.ai.entity.made.RocketBrokerMessage;
import com.ai.service.made.IRocketBrokerMessageService;
import com.ai.service.rocketMq.RocketMQProducerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.Objects;
import java.util.UUID;

/**
 * @Author jiaok
 * @Date 2023-11-08 14:53
 */
@Slf4j
@Component
public class RocketMQProducerServiceImpl implements RocketMQProducerService {


    private static final String topic = "THREE_STEEL_VIDEO_ANALYSIS_ALARM_TOPIC";
    @Autowired
    private IRocketBrokerMessageService iRocketBrokerMessageService;
    private final Long asyncFailTime=3000L;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void asyncSendMsg(String msgBody,String tag,String alarmRecordID) throws RuntimeException{

        String messageId = UUID.randomUUID().toString().replaceAll("-", "");
        log.info("===========================  messageId:{} ===========================",messageId);

        RocketBrokerMessage rocketBrokerMessage = new RocketBrokerMessage();
        rocketBrokerMessage.setId(messageId);
        rocketBrokerMessage.setAlarmRecordId(alarmRecordID);
        rocketBrokerMessage.setMessage(msgBody);
        rocketBrokerMessage.setCreateTime(new SimpleDateFormat("yyyy--MM--dd HH:mm:ss").format(new Date()).toString());
        rocketBrokerMessage.setRetryTimes(0);

        Message<String> message = MessageBuilder
                .withPayload(msgBody)
                //指定key
                .setHeader("KEYS",messageId )
                .build();

        rocketMQTemplate.asyncSend(topic+":"+tag, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("=========================== # RocketMQProducerServiceImpl.sendAsyncMsg # 消息投递：{} ==========================="
                        , sendResult.getSendStatus());
                rocketBrokerMessage.setMessageStatus(MqMessageStatus.SENDING_SUCCESS.getCode());
                iRocketBrokerMessageService.saveOrUpdate(rocketBrokerMessage);
            }

            @Override
            public void onException(Throwable throwable) {
                rocketBrokerMessage.setMessageStatus(MqMessageStatus.SENDING_EXCEPTION.getCode());
                iRocketBrokerMessageService.saveOrUpdate(rocketBrokerMessage);
                log.error(throwable.getMessage());
                log.error(Arrays.toString(throwable.getStackTrace()));
            }
        }, asyncFailTime);

    }

    public void synSendMsg(String msgBody,String tag) {

        rocketMQTemplate.asyncSend(topic+":"+tag, MessageBuilder.withPayload(msgBody).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("=========================== # RocketMQProducerServiceImpl.sendAsyncMsg # 消息投递：{} ==========================="
                        , sendResult.getSendStatus());
                log.info("=========================== SendMessageBody:{}  ===========================",msgBody);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error(throwable.getMessage());
                log.error(Arrays.toString(throwable.getStackTrace()));
            }
        }, asyncFailTime);
    }
}
